package org.seed.rpc.serverstub;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.seed.rpc.common.codec.RpcDecoder;
import org.seed.rpc.common.codec.RpcEncoder;
import org.seed.rpc.common.io.RpcRequest;
import org.seed.rpc.common.io.RpcResponse;
import org.seed.rpc.common.util.IconSpecialUtil;
import org.seed.rpc.core.RegistryClient;
import org.seed.rpc.core.RegistryOperator;
import org.seed.rpc.core.RpcService;
import org.seed.rpc.core.RuntimeLocal;
import org.seed.rpc.core.etcd.EtcdClient;
import org.seed.rpc.serverstub.health.RpcServiceHealthImpl;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.util.Assert;

import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.*;


/**
 * 启动 netty-server 监听端口，并将使用 @RpcService 标记的注解注册到 spring 容器
 */
@Slf4j
@EnableConfigurationProperties(RpcConfigProperties.class)
@Configuration
@Import(RpcServiceHealthImpl.class)
@ConditionalOnProperty("rpc.local.registry")
public class RpcServerAutoConfigure implements ApplicationRunner, ApplicationContextAware, DisposableBean {

    /**
     * 存放 「服务名」与「服务对象」的映射关系
     */
    private Map<String, Object> handlerMap = new HashMap<>();

    private ApplicationContext applicationContext;

    private RpcConfigProperties rpcServerProperties;

    /**
     * netty-server的「bossGroup」
     */
    private EventLoopGroup bossGroup;
    /**
     * netty-server的「workerGroup」
     */
    private EventLoopGroup workerGroup;
    /**
     * netty-server的「channelFuture」
     */
    private ChannelFuture channelFuture;

    private RegistryClient etcdClient;

    public RpcServerAutoConfigure(@Autowired RpcConfigProperties rpcServerProperties) {
        this.rpcServerProperties = rpcServerProperties;
    }


    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 1. 收集spring容器的bean对象, 存储至本地映射
        gatherRpcServiceObject();
        // 2. 启动rpc-server的socket监听
        if (handlerMap.size() == 0) {
            return;
        }
        boolean started = listenAndServeRpcSocket();
        if (!started) {
            log.error("\uD83D\uDE26 \uD83D\uDE26 \uD83D\uDE26 rpc-server started failed.");
            return;
        }
        // 3. 将本地RPC服务注册至注册中心
        RegistryClient registryClient = RuntimeLocal.getRegistryClient();
        if (registryClient == null) {
            etcdClient = new EtcdClient(rpcServerProperties.getRegistry());
            RuntimeLocal.setRegistryClient(etcdClient);
        } else {
            etcdClient = registryClient;
        }
        try {
            publishToRegistry();
            log.info("\uD83C\uDF1E \uD83C\uDF1E \uD83C\uDF1E rpc-server started succeed.");
        } catch (Exception e) {
            log.error("\uD83D\uDE28 \uD83D\uDE28 \uD83D\uDE28 etcd-server register error: {}", e.getMessage());
            e.printStackTrace();
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    /**
     * 在 spring factory 中查找服务对象
     */
    private void gatherRpcServiceObject() {
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(RpcService.class);
        if (beans == null || beans.size() == 0) {
            return;
        }
        if (beans.size() > 1) {
            log.info("╔══════════════════════════════════════════════════════════════════════════════════════════════════►");
        }
        for (Object value : beans.values()) {
            Class[] interfaces = value.getClass().getInterfaces();
            String interfaceName = null;
            for (Class clazz : interfaces) {
                if (clazz.isAnnotationPresent(RpcService.class)) {
                    interfaceName = clazz.getName();
                    break;
                }
            }
            handlerMap.put(interfaceName, value);
            if (!"org.seed.rpc.serverstub.health.RpcServiceHealth".equals(interfaceName)) {
                log.info("║ ➜ {} ➫ {}", value.getClass().getName(), interfaceName);
            }

        }
        if (beans.size() > 1) {
            log.info("╚══════════════════════════════════════════════════════════════════════════════════════════════════►");
        }

    }


    /**
     * 开启 rpc-socket 监听
     */
    private boolean listenAndServeRpcSocket() {
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            bossGroup = new NioEventLoopGroup();
            workerGroup = new NioEventLoopGroup();
            serverBootstrap.group(bossGroup, workerGroup);
            serverBootstrap.channel(NioServerSocketChannel.class);
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel channel) throws Exception {
                    ChannelPipeline pipeline = channel.pipeline();
                    pipeline.addLast(new RpcDecoder(RpcRequest.class));   // 解码 RPC 请求
                    pipeline.addLast(new RpcEncoder(RpcResponse.class));  // 编码 RPC 响应
                    pipeline.addLast(new RpcServerHandler(handlerMap));   // 处理 RPC 请求
                }
            });

            serverBootstrap.option(ChannelOption.SO_BACKLOG, 1024);
            serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
            // 获取 RPC 服务器的 IP 地址与端口号
            String ip = rpcServerProperties.getIp();
            if (Objects.isNull(ip) || "".equals(ip)) {
                ip = getEth0Ip();
            }
            Integer port = rpcServerProperties.getPort();
            if (Objects.isNull(port) || port.intValue() == 0) {
                port = 31792;
            }
            // 启动 RPC 服务器
            channelFuture = serverBootstrap.bind(ip, port).sync();
            log.info("rpc-socket is listen and serve on port: {}, local ip:{}", port, ip);
            return true;
        } catch (Exception e) {
            log.error(e.getMessage());
            e.printStackTrace();
            destroy();
            return false;
        }
    }


    /**
     * 发布本地注册到注册中心
     */
    private void publishToRegistry() throws Exception {
        Assert.hasText(rpcServerProperties.getServiceName(), "has not set service-name or application-name yet.");
        RegistryOperator.publishLocalEndpoint(rpcServerProperties.getServiceName(), rpcServerProperties.getIp() + ":" + rpcServerProperties.getPort(), etcdClient);
        log.info(IconSpecialUtil.randomIcon() + " registered to rpc-registry success.");
    }

    private boolean unmountFromRegistry() {
        try {
            RegistryOperator.unloadLocalEndpoint(rpcServerProperties.getServiceName(), rpcServerProperties.getIp() + ":" + rpcServerProperties.getPort(), etcdClient);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    @Override
    public void destroy() {
        // 1. 从注册中心卸载
        unmountFromRegistry();

        // 2. 关闭本地netty服务线程
        try {
            if (channelFuture != null) {
                channelFuture.channel().closeFuture().sync();
            }
        } catch (Exception e) {
            log.debug("netty-channel closed.");
        }
        try {
            if (bossGroup != null) {
                bossGroup.shutdownGracefully();
            }
        } catch (Exception e) {
            log.debug("netty-boss-group closed.");
        }
        try {
            if (workerGroup != null) {
                workerGroup.shutdownGracefully();
            }
        } catch (Exception e) {
            log.debug("netty-worker-group closed.");
        }
    }


    /**
     * 查询 eth0 网卡的 ip 地址
     *
     * <p></p>
     */
    private String getEth0Ip() {
        Enumeration<NetworkInterface> nifs = null;
        try {
            nifs = NetworkInterface.getNetworkInterfaces();
        } catch (SocketException e) {
            e.printStackTrace();
            return "127.0.0.1";
        }
        while (nifs.hasMoreElements()) {
            NetworkInterface nif = nifs.nextElement();
            Enumeration<InetAddress> addresses = nif.getInetAddresses();
            while (addresses.hasMoreElements()) {
                InetAddress inetAddress = addresses.nextElement();
                if (inetAddress instanceof Inet4Address) {
                    if ("eth0".equalsIgnoreCase(nif.getName())) {
                        return inetAddress.getHostAddress();
                    }
                }
            }
        }
        return "127.0.0.1";
    }

}
